package gmall;

import demo.MyInterceptor;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
                                // flume 采集日志信息到 kafka
public class GmallInterceptor implements Interceptor {
    @Override
    public void initialize() {
        // 不用管
    }

    @Override
    public Event intercept(Event event) {
        // 1、明确需求：怎么打标记（启动日志 标记 log_type:start  行为日志 标记 log_type:event）
        String str = new String(event.getBody());
        // 2、识别数据，将body数据转为string，识别其中是否包含 "en":"start"
        // 定义一个flag,用作标识数据是否正常，如果不正常则返回false,此时输出(return) null
        boolean flag = true;
        if (str.contains("\"en\":\"start\"")) {
            // 3、针对不同的数据进行标记（标记的同时，要识别数据是否正常）
            //   启动日志 需要是一个json字符串，最简单的方法就是首尾是否是花括号
            event.getHeaders().put("log_type","start");
            flag = ETLUtile.checkStart(str);
        }else {
            //   行为日志 需要有两个元素：一个是时间戳，一个是json字符串
            event.getHeaders().put("log_type","event");
            flag = ETLUtile.checkEvent(str);
        }

        if (flag){
            // 输出加工好的event
            return event;
        }
        // 4、输出加工好的event
        return null;
    }


    // 为了提升效率，返回对象 retEvents 创建为全局变量
     ArrayList<Event> retEvents = new ArrayList<>();
    @Override
    public List<Event> intercept(List<Event> list) {
        // 我们在一个拦截器中写的内容是可能会输出null值的，处理后的list中可能会出现null值
        // 不利于我们的后续处理，所以我们要清除这个null （ null值不输出 ）
        retEvents.clear();  // 每一次成批处理数据时，要先清空再插入
        for (Event event : list) {
            retEvents.add(intercept(event));
        }
        return null;
    }

    @Override
    public void close() {
        // 不用管
    }

    public static class Builder implements Interceptor.Builder{

        @Override
        public Interceptor build() {
            return new GmallInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
